package com.redis.queue;

import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.stream.StreamListener;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

/**
 * 自适应监听器。
 *      redis stream 消息队列 实现类。
 *          提供消息转换方法 可以将消息体转换为对象。基础并重写此方法 可以对消息进行业务操作.
 * @param <T>
 */
@Slf4j
public abstract class AdaptiveListener<T extends Message> implements StreamListener<String, ObjectRecord<String, String>> {


    private RedisQueueUtils redisQueueUtils;

    private  Class<?> tClass = currentModelClass(getClass());

    public abstract void consumer(T t);

    @SneakyThrows
    @Override
    public void onMessage(ObjectRecord<String, String> objectRecord) {
        log.info("进入方法得到数据：{}",objectRecord);
        Object o = JSON.parseObject(objectRecord.getValue(), tClass);
        consumer((T) o);
        //消费完毕删除该条消息
        redisQueueUtils.ack(objectRecord.getStream(),CacheConstant.MESSAGE_GROUP, objectRecord.getId().getValue());
        redisQueueUtils.del(objectRecord.getStream(), objectRecord.getId().getValue());
    }


    /**
     * 获取泛型对象
     *
     * @param clazz
     * @return
     */
    public static Class<?> currentModelClass(Class<?> clazz) {
        // 获取超类
        Type type = clazz.getGenericSuperclass();
        if (!(type instanceof ParameterizedType)) {
            throw new RuntimeException("超类没有参数化类型");
        }
        // 获取超累泛型上设置的类
        Type trueType = ((ParameterizedType) type).getActualTypeArguments()[0];
        if (!(trueType instanceof Class)) {
            throw new RuntimeException("没有在超累泛型上设置实际的类");
        }

        return (Class<?>) trueType;
    }


    public RedisQueueUtils getRedisQueueUtils() {
        return redisQueueUtils;
    }

    public void setRedisQueueUtils(RedisQueueUtils redisQueueUtils) {
        this.redisQueueUtils = redisQueueUtils;
    }
}

